package org.yuanzheng.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

import scala.util.Random

/**
 * @author yuanzheng
 * @date 2020/6/12-13:30
 */
class CustomSource extends SourceFunction[StationLog] {
  var flag = true;

  //启动一个source并从source中获取数据，run方法停止则数据流停止
  override def run(sourceContext: SourceFunction.SourceContext[StationLog]): Unit = {
    //StationLog(sid: String, callOut: String, callIn: String, callType: String, callTime: Long, duration: Long)
    val r = new Random()
    val types = Array("fail", "busy", "refuse", "success")
    while (flag) {
      1.to(10).map(i => {
        var callOut = "1571386%04d".format(r.nextInt(10000))
        var callIn = "1863957%04d".format(r.nextInt(10000))
        new StationLog("Sid" + r.nextInt(10), callOut, callIn, types(r.nextInt(4)), System.currentTimeMillis(), r.nextInt(20))
      }).foreach(sourceContext.collect(_))
      Thread.sleep(2000)
    }
  }

  // 终止数据流
  override def cancel(): Unit = {
    flag = false;
  }
}

object UserDefinedSource {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    import org.apache.flink.streaming.api.scala._
    val stream: DataStream[StationLog] = environment.addSource(new CustomSource)
    stream.print()
    environment.execute()
  }
}
